package com.zsy.rocketdemo.produce;

import com.alibaba.fastjson.JSON;
import lombok.Getter;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.apache.rocketmq.remoting.exception.RemotingException;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.io.UnsupportedEncodingException;
import java.util.stream.IntStream;

@Service
public class ZsyTransProducer{

    private TransactionMQProducer transactionMQProducer;

    @Resource(name = "zsyTransListener")
    private TransactionListener transactionListener;

    @Getter
    private String topic = "zsyTopicTest";
    @Getter
    protected String tag;

    @PostConstruct
    public void init() throws MQClientException {
        tag = "zsyTransProducer";
        transactionMQProducer = new TransactionMQProducer(this.tag);
        transactionMQProducer.setNamesrvAddr("47.100.22.19:9876");
        transactionMQProducer.setInstanceName(this.tag);
        transactionMQProducer.setTransactionListener(transactionListener);
        transactionMQProducer.start();
        System.out.println(this.tag + "Producer启动。");
    }

    public void sendMessage(Message message) throws InterruptedException, RemotingException, MQClientException, MQBrokerException {
        SendResult sendResult = transactionMQProducer.sendMessageInTransaction(message, null);
        System.out.println(JSON.toJSONString(sendResult));
    }

    public void sendMessage(int num) {
        IntStream.range(0, num).forEach(i -> {
            try {
                sendMessage(buildMessage(i));
            } catch (MQClientException | RemotingException | MQBrokerException | InterruptedException | UnsupportedEncodingException e) {
                e.printStackTrace();
            }
        });
    }

    public Message buildMessage(int num) throws UnsupportedEncodingException {
        return new Message(getTopic(),
                getTag(),
                ("zsy Hello RocketMQ " + getTag() + " " + num).getBytes(RemotingHelper.DEFAULT_CHARSET)
        );
    }
}
